package org.yray.io.rocketmqdemo.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.yray.io.rocketmqdemo.listener.MsgDispatcher;

//@Configuration
@Slf4j
public class MqConfig {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.group}")
    private String consumerGroup;
    @Value("${rocketmq.name-server}")
    private String namesrvaddr;

    @Autowired
    MsgDispatcher msgDispatcher;

    //这里获取所有服务以不同tags去作分发,如果是重复发送,如果是相同groupName instanceName并需设为不同
    //tags当作消费监听器类型
    //如果是不同groupName会导致消息重复消费
    @Bean(name = "normalMqConsumer")
    public DefaultMQPushConsumer initNormalMqConsumer() {
        log.info("init defaultMQConsumer");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvaddr);
        String instanceName = RandomUtils.nextLong()+"";
        consumer.setInstanceName(instanceName);
        String topic="my-topic";
        try {
            consumer.subscribe(topic,"*");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener((MessageListenerConcurrently)(list, context)->{
                ConsumeConcurrentlyStatus result = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                try {
                    for (MessageExt messageExt : list) {
                        String msg=new String(messageExt.getBody());
                        result= MsgDispatcher.getMqMsgListener(messageExt.getTags()).doMsg(msg);
                        log.info("消费消息：{},tags={},topic={}",msg,messageExt.getTags(),messageExt.getTopic());
                    }
                } catch (Exception e) {
                    log.error("启动监听MQ失败{}", ExceptionUtils.getMessage(e));
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//稍后再试
                }
                return result;//消费成功
            });
            consumer.start();
        } catch (MQClientException e) {
            log.error("启动监听MQ失败{}", ExceptionUtils.getMessage(e));
        }
        log.info("init defaultMQConsumer succ,instanceName={},group={},topic={}",instanceName,consumerGroup,topic);
        return consumer;
    }



}
